package com.zhanghe.study;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Printed;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;

/**
 * 测试kstream
 */
public class KStreamTest {
    public static void main(String[] args) {
        Properties prop = init();
        StreamsBuilder builder = new StreamsBuilder();
        // 构造kstream日志流（数据来源来自  kafka的主题stream-input）
        KStream<String,String> source = builder.stream("stream-input");
        // 打印到控制台
        source.print(Printed.<String, String>toSysOut());
        Topology topology = builder.build();
        System.out.println(topology.describe());
        final KafkaStreams kafkaStreams = new KafkaStreams(topology,prop);
        final CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread("hook"){
            @Override
            public void run() {
                System.out.println("线程停止");
                kafkaStreams.close();
                latch.countDown();
            }
        });
        try {
            kafkaStreams.start();
            latch.await();
        } catch (InterruptedException e){
            System.exit(1);
        }
        System.out.println("项目终止");
        System.exit(0);


    }

    public static Properties init(){
        Properties properties = new Properties();
        // 流处理应用的id，必须指定
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"KStream-test");
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        // key序列化反序列化
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        //value序列化反序列化
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        return properties;
    }
}
